package com.eshipenet.shipnet.dp_decoder

import java.lang

import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema
import org.apache.kafka.clients.producer.ProducerRecord
import org.json4s.native.Serialization.write
import org.json4s.{DefaultFormats, Formats}

object EdgeChannelDataPacketSerializer {
  implicit lazy val formats: Formats = DefaultFormats
}

class EdgeChannelDataPacketSerializer extends KafkaSerializationSchema[EdgeChannelDataPacket] {
  import EdgeChannelDataPacketSerializer._

  override def serialize(element: EdgeChannelDataPacket,
                         timestamp: lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
    val topic = "EdgeChannelDataPacketDecoded2"
    val key   = element.id.getBytes
    val obj   = element.aux.values ++ Map("id" -> element.id, "data" -> element.data)
    val value = write(obj).getBytes
    new ProducerRecord(topic, key, value)
  }

}
